﻿
#include <doctest/doctest.h>
#include "log.h"

#include <thread>
/*
 *
 *  缩略语	英语	                    中文	        解析
 *  SQ	    Submission Queue	    提交队列	    一整块连续的内存空间存储的环形队列。 用于存放将执行操作的数据。
 *  CQ	    Completion Queue	    完成队列	    一整块连续的内存空间存储的环形队列。 用于存放完成操作返回的结果。
 *  SQE	    Submission Queue Entry	提交队列项	提交队列中的一项。
 *  CQE	    Completion Queue Entry	完成队列项	完成队列中的一项。
 *  Ring	Ring	                环	        比如 SQ Ring，就是“提交队列信息”的意思。 包含队列数据、队列大小、丢失项等等信息。
 *
 *  SQ 和 CQ 中每个节点保存的都是他们在 SQEs 数组的偏移量，实际的请求只保存在 SQEs 数组中
 *
 *SQE：
 * Opcode：      描述要进行的系统调用的IO 操作码。如果是读，操作码IORING_OP_READV。
 * Flags：       修饰符，可以通过任何请求传递
 * Fd：          要读取的文件描述符
 * Address：     对于我们的readv调用，它创建了一个缓冲区(或向量)数组来读入数据。因此，address字段包含了该数组的地址。
 * Length：      Address 缓冲区 向量数组的长度。
 * User Data：   通常这是一个指针，指向一些结构体，其中保存了请求的元数据，来识别应用的请求。
 *              当请求从CQ 队列中出来时，并不能保证IO结果与 请求SQEs的顺序相同。
 *              如果一定保证有序的就会降低性能， 就违背了异步API的初衷。
 *              因此，我们需要一些东西来识别我们发出的请求。
 *              User Data这可以达到这个目的。
 *
 * CQE:
 *  Result：      readv系统调用的返回值。如果成功，就会有读取的字节数; 否则它将有一个错误代码。
 *  User Data：   在SQE中传递的指针。
 *
 *
 * io_uring在准备阶段，会涉及到三个系统调用:
 *
 * io_uring_setup       :     返回值是一个 fd，应用程序使用这个 fd 进行 mmap，和 kernel 共享一块内存
                                entries: 要创建的sqe的数量
                                params: 用户层指定的参数




 * io_uring_enter       :
 * io_uring_register    :
 *
 *
 * io_sqring_offset     :   表示 SQ 的指针在 mmap 中的 offset
 * io_cqring_offset     :   表示 CQ 的指针在 mmap 中的 offset
 *
 *
 *
 *  io_uring_create:
 *
 *
 *
 * flags只支持这些标志，如果有其它标志都会报错
	    #define IORING_SETUP_IOPOLL	(1U << 0)	// io poll 模式
		#define IORING_SETUP_SQPOLL	(1U << 1)	// sq poll 模式
		#define IORING_SETUP_SQ_AFF	(1U << 2)	// 指定线程cpu时指定这个参数
		#define IORING_SETUP_CQSIZE	(1U << 3)	// 应用设置完成队列大小
		#define IORING_SETUP_CLAMP	(1U << 4)	// 当用户指定的entries太大时，可以把值改小
		#define IORING_SETUP_ATTACH_WQ	(1U << 5)	//添加到当前已经存在的wq里
		#define IORING_SETUP_R_DISABLED	(1U << 6)	// 如果是sq-poll模式，一开始不启动sq-thread
 */
#include <stdio.h>
#include <fcntl.h>
#include <string.h>
#include <stdlib.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include "liburing.h"


#include <boost/asio/io_service.hpp>
#include <boost/thread/thread.hpp>
#include <boost/bind.hpp>
#include <iostream>

#define BOOST_ASIO_HAS_CO_AWAIT

#include <filesystem>
#include <thread>

#include <coroutine>
//#include "asio.hpp"
#include <boost/asio.hpp>
#include <boost/asio/co_spawn.hpp>
#include <boost/asio/detached.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/signal_set.hpp>
#include <boost/asio/write.hpp>
#include <boost/asio/io_service.hpp>
#include <boost/asio/awaitable.hpp>
#include <boost/asio/awaitable.hpp>
#include <boost/asio/buffer.hpp>
#include <boost/asio/co_spawn.hpp>
#include <boost/asio/connect.hpp>
#include <boost/asio/detached.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/read.hpp>
#include <boost/asio/signal_set.hpp>
#include <boost/asio/this_coro.hpp>
#include <boost/asio/use_awaitable.hpp>
#include <boost/asio/write.hpp>
#include "gtest/gtest.h"
#include "log.h"

using boost::asio::awaitable;
using boost::asio::co_spawn;
using boost::asio::detached;
using boost::asio::use_awaitable;
using boost::asio::ip::tcp;
namespace this_coro = boost::asio::this_coro;


#if defined(BOOST_ASIO_ENABLE_HANDLER_TRACKING)
# define use_awaitable \
  boost::asio::use_awaitable_t(__FILE__, __LINE__, __PRETTY_FUNCTION__)
#endif

using namespace boost;
using namespace boost::asio;

#define QD    4 // io_uring 队列长度

TEST_CASE("modern_features_test01") {
    int i, fd, pending, done, ret, offset;
    void *buf;

    // 1. 初始化一个 io_uring 实例
    struct io_uring ring;
    ret = io_uring_queue_init(QD,    // 队列长度
                              &ring, // io_uring 实例
                              0);    // flags，0 表示默认配置，例如使用中断驱动模式

    const char *file_name = "./stddef.h";
    // 2. 打开输入文件，注意这里指定了 O_DIRECT flag，内核轮询模式需要这个 flag，见前面介绍
    fd = open(file_name, O_RDONLY | O_DIRECT);
    struct stat sb;
    fstat(fd, &sb); // 获取文件信息，例如文件长度，后面会用到

    // 3. 初始化 4 个读缓冲区
    ssize_t fsize = 0;             // 程序的最大读取长度
    struct iovec *iovecs = (iovec *) calloc(QD, sizeof(struct iovec));
    for (i = 0; i < QD; i++) {
        if (posix_memalign(&buf, 4096, 4096))
            return;
        iovecs[i].iov_base = buf;  // 起始地址
        iovecs[i].iov_len = 4096;  // 缓冲区大小
        fsize += 4096;
    }

    // 4. 依次准备 4 个 SQE 读请求，指定将随后读入的数据写入 iovecs
    struct io_uring_sqe *sqe;
    offset = 0;
    i = 0;
    do {
        sqe = io_uring_get_sqe(&ring);  // 获取可用 SQE
        io_uring_prep_readv(sqe,        // 用这个 SQE 准备一个待提交的 read 操作
                            fd,         // 从 fd 打开的文件中读取数据
                            &iovecs[i], // iovec 地址，读到的数据写入 iovec 缓冲区
                            1,          // iovec 数量
                            offset);    // 读取操作的起始地址偏移量
        offset += iovecs[i].iov_len;    // 更新偏移量，下次使用
        i++;

        if (offset > sb.st_size)        // 如果超出了文件大小，停止准备后面的 SQE
            break;
    } while (1);

    // 5. 提交 SQE 读请求
    ret = io_uring_submit(&ring);       // 4 个 SQE 一次提交，返回提交成功的 SQE 数量
    if (ret < 0) {
        fprintf(stderr, "io_uring_submit: %s\n", strerror(-ret));
        return;
    } else if (ret != i) {
        fprintf(stderr, "io_uring_submit submitted less %d\n", ret);
        return;
    }

    // 6. 等待读请求完成（CQE）
    struct io_uring_cqe *cqe;
    done = 0;
    pending = ret;
    fsize = 0;
    for (i = 0; i < pending; i++) {
        io_uring_wait_cqe(&ring, &cqe);  // 等待系统返回一个读完成事件
        done++;

        if (cqe->res != 4096 && cqe->res + fsize != sb.st_size) {
            fprintf(stderr, "ret=%d, wanted 4096\n", cqe->res);
        }

        fsize += cqe->res;
        io_uring_cqe_seen(&ring, cqe);   // 更新 io_uring 实例的完成队列
    }

    // 7. 打印统计信息
    printf("Submitted=%d, completed=%d, bytes=%lu\n", pending, done, (unsigned long) fsize);

    // 8. 清理工作
    close(fd);
    io_uring_queue_exit(&ring);

}

boost::mutex global_stream_lock;

void WorkerThread(boost::shared_ptr<boost::asio::io_service> io_service) {
    global_stream_lock.lock();
    std::cout << "[" << boost::this_thread::get_id() << "] Thread Start" << std::endl;
    global_stream_lock.unlock();

    io_service->run();

    global_stream_lock.lock();
    std::cout << "[" << boost::this_thread::get_id() << "] Thread Finish" << std::endl;
    global_stream_lock.unlock();
}

void Dispatch(int x) {
    global_stream_lock.lock();
    std::cout << "[" << boost::this_thread::get_id() << "] " << __FUNCTION__ << " x = " << x << std::endl;
    global_stream_lock.unlock();
}

void Post(int x) {
    global_stream_lock.lock();
    std::cout << "[" << boost::this_thread::get_id() << "] " << __FUNCTION__ << " x = " << x << std::endl;
    global_stream_lock.unlock();
}

void Run3(boost::shared_ptr<boost::asio::io_service> io_service) {
    for (int x = 0; x < 3; ++x) {
        io_service->dispatch([capture0 = x * 2] { return Dispatch(capture0); });
        io_service->post([capture0 = x * 2 + 1] { return Post(capture0); });
        boost::this_thread::sleep(boost::posix_time::milliseconds(1000));
    }
}

TEST_CASE("modern_features_asio_post")
{
    boost::shared_ptr<boost::asio::io_service> io_service(new boost::asio::io_service);
    boost::shared_ptr<boost::asio::io_service::work> work(new boost::asio::io_service::work(*io_service));

    global_stream_lock.lock();
    std::cout << "[" << boost::this_thread::get_id() << "] The program will exit when all  work has finished."
              << std::endl;
    global_stream_lock.unlock();

    boost::thread_group worker_threads;
    for (int x = 0; x < 1; ++x) {
        worker_threads.create_thread([io_service] { return WorkerThread(io_service); });
    }

    io_service->post([io_service] { return Run3(io_service); });

    work.reset();

    worker_threads.join_all();
}


using namespace asio::ip;


awaitable<void> echo(tcp::socket socket) {
    try {
        char data[1024];
        for (;;) {
            std::size_t n = co_await socket.async_read_some(boost::asio::buffer(data), use_awaitable);
            co_await async_write(socket, boost::asio::buffer(data, n), use_awaitable);
        }
    }
    catch (std::exception &e) {
        std::printf("echo Exception: %s\n", e.what());
    }
}

awaitable<void> listener() {
    auto executor = co_await this_coro::executor;
    tcp::acceptor acceptor(executor, {tcp::v4(), 8082});
    for (;;) {
        tcp::socket socket = co_await acceptor.async_accept(use_awaitable);
        co_spawn(executor, echo(std::move(socket)), detached);
    }
}

TEST_CASE("coroutines_asio_cat") {
    {
        try {
            boost::asio::io_context io_context(1);

            boost::asio::signal_set signals(io_context, SIGINT, SIGTERM);
            signals.async_wait([&](auto, auto) { io_context.stop(); });

// 执行协程，并设置了一个callback函数来获取这个协程的返回值
            co_spawn(io_context, listener(), detached);

            io_context.run();
        }
        catch (std::exception &e) {
            std::printf("Exception: %s\n", e.what());
        }
    }
}

#include <arpa/inet.h>
/*#include <bits/stdc++.h>*/
#include <liburing.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <unistd.h>

const int BUFSIZE = 1024;
struct request {
    enum STATE {
        ACCEPT, READ, WRITE
    };
    int fd;
    STATE state;
    union {
        struct {
            sockaddr_in ipv4_addr;
            socklen_t lens;
        } addr;
        char buf[BUFSIZE];
    };
};

class IOuring {
    io_uring ring;

public:
    IOuring(int queue_size) { io_uring_queue_init(queue_size, &ring, 0); }

    ~IOuring() { io_uring_queue_exit(&ring); }

    void seen(io_uring_cqe *cqe) { io_uring_cqe_seen(&ring, cqe); }

    int wait(io_uring_cqe **cqe) { return io_uring_wait_cqe(&ring, cqe); }

    int submit() { return io_uring_submit(&ring); }

    void accpet_asyn(int sock_fd, request *body) {
        auto sqe = io_uring_get_sqe(&ring);
        body->state = request::ACCEPT;
        body->fd = sock_fd;
        body->addr.lens = sizeof(sockaddr_in);
        io_uring_prep_accept(sqe, sock_fd, (sockaddr *) &(body->addr.ipv4_addr),
                             &(body->addr.lens), 0);
        io_uring_sqe_set_data(sqe, body);
    }

    void read_asyn(int client_fd, request *body) {
        auto sqe = io_uring_get_sqe(&ring);
        body->state = request::READ;
        body->fd = client_fd;
        io_uring_prep_read(sqe, client_fd, body->buf, sizeof(body->buf), -1);
        io_uring_sqe_set_data(sqe, body);
    }

    void write_asyn(int client_fd, request *body) {
        auto sqe = io_uring_get_sqe(&ring);
        body->state = request::WRITE;
        body->fd = client_fd;
        io_uring_prep_write(sqe, client_fd, body->buf, sizeof(body->buf), -1);
        io_uring_sqe_set_data(sqe, body);
    }
};

TEST_CASE("liburing_test") {
    /*init socket*/
    int sock_fd = socket(AF_INET, SOCK_STREAM, 0);
    sockaddr_in sock_addr;
    sock_addr.sin_port = htons(8000);
    sock_addr.sin_family = AF_INET;
    sock_addr.sin_addr.s_addr = INADDR_ANY;
    int ret = ::bind(sock_fd, (sockaddr *) &sock_addr, sizeof(sock_addr));
    perror("");
    listen(sock_fd, 10);

    std::cout << "listen begin ..." << std::endl;

    /*io_uring*/
    IOuring ring(1024);

    ring.accpet_asyn(sock_fd, new request);
    ring.submit();

    while (true) {
        io_uring_cqe *cqe;
        ring.wait(&cqe);
        request *res = (request *) cqe->user_data;
        switch (res->state) {
            case request::ACCEPT:
                if (cqe->res > 0) {
                    int client_fd = cqe->res;
                    ring.accpet_asyn(sock_fd, res);
                    ring.read_asyn(client_fd, new request);
                    ring.submit();
                }
                std::cout << "client_fd" << cqe->res << std::endl;
                break;
            case request::READ:
                if (cqe->res > 0) std::cout << res->buf << std::endl;
                ring.write_asyn(res->fd, res);
                ring.submit();
                break;
            case request::WRITE:
                if (cqe->res > 0) {
                    close(res->fd);
                    delete res;
                }
                break;
            default:
                std::cout << "error " << std::endl;
                break;
        }
        ring.seen(cqe);
    }
}

void thread_1() {
    std::cout << "子线程1" << std::endl;
    usleep(10000);
}

void thread_2(const int &x) {
    std::cout << "x:" << x << std::endl;
    std::cout << "子线程2" << std::endl;
}

TEST_CASE("test_thread") {
    std::thread first(thread_1); // 开启线程，调用：thread_1()
    std::thread second(thread_2, 100); // 开启线程，调用：thread_2(100)
//thread third(thread_2,3);//开启第3个线程，共享thread_2函数。
    std::cout << "主线程\n";

    first.join(); //必须说明添加线程的方式
    second.join();
    std::cout << "子线程结束.\n";//必须join完成
}